package com.Reactor3参考文档;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;

import org.junit.Test;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

public class DoFinally {


	@Test
	// 使用资源和 try-catch 代码块
	// 最后一个要与命令式编程对应的对比就是使用 Java 7 "try-with-resources" 或 finally 代码块清理资源。在 Reactor 中都有对应的方法： using 和 doFinally
	public void test8(){
		AtomicBoolean isDisposed = new AtomicBoolean();
		Disposable disposableInstance = new Disposable() {
		    @Override
		    public void dispose() {
		        isDisposed.set(true);
		    }

		    @Override
		    public String toString() {
		        return "DISPOSABLE";
		    }
		};

		Flux<String> flux =
		Flux.using(
				// 第一个 lambda 生成资源，这里我们返回模拟的（mock） Disposable
		        () -> disposableInstance,
		        // 第二个 lambda 处理资源，返回一个 Flux<T>。
		        disposable -> Flux.just(disposable.toString()),
		        // 第三个 lambda 在 2) 中的资源 Flux 终止或取消的时候，用于清理资源
		        Disposable::dispose
		);
		// 在订阅或执行流序列之后， isDisposed 会置为 true
	}

	@Test
	// 另一方面， doFinally 在序列终止（无论是 onComplete、`onError`还是取消）的时候被执行， 并且能够判断是什么类型的终止事件（完成、错误还是取消？）
	public void test9(){
		// 我们想进行统计，所以用到了 LongAdder
		LongAdder statsCancel = new LongAdder();

		Flux<String> flux =
		Flux.just("foo", "bar")
		    .doFinally(type -> {
		    	// doFinally 用 SignalType 检查了终止信号的类型，如果只是取消，那么统计数据自增
		        if (type == SignalType.CANCEL)
		          statsCancel.increment();
		    })
		    // take(1) 能够在发出 1 个元素后取消流
		    .take(1);
	}

}
